package com.nami.logic.forward.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import com.nami.basic.bo.servicetrans.up.UpMsgWrap;
import com.nami.basic.constants.DownTopicConst;
import com.nami.basic.enums.message.MsgTypeEnum;
import com.nami.basic.enums.protocol.DownCMDEnum;
import com.nami.basic.packet.Packet;
import com.nami.basic.packet.message.Message;
import com.nami.gate.api.request.message.DownMsgBatchReq;
import com.nami.gate.api.request.message.DownMsgRequest;
import com.nami.logic.bo.send.DownModeConfig;
import com.nami.logic.bo.send.PacketTransBO;
import com.nami.logic.enums.DeviceEnum;
import com.nami.logic.forward.TransDownMsgHandler;
import com.nami.logic.property.DownModeProperty;
import com.nami.logic.qos.QoSReceiveDaemonC2S;
import com.nami.logic.qos.QosSendDaemonS2C;
import com.nami.logic.service.DownMsgSender;
import com.nami.logic.service.PacketFactory;
import com.nami.logic.service.UserService;
import com.nami.store.api.bo.UserDevice;
import com.nami.store.api.bo.UserDeviceDetailInfo;
import com.nami.store.api.bo.UserWrap;
import com.nami.store.api.response.groupmember.GroupMemberDTO;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.dromara.common.core.enums.conver.ConverTypeEnum;
import org.dromara.common.core.enums.message.MessageFlowEnum;
import org.dromara.common.core.utils.IdentityUtil;
import org.dromara.common.json.utils.JsonUtils;
import org.jetbrains.annotations.NotNull;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;

import static com.nami.basic.constants.CommonConst.COLON;

@Slf4j
@Service
@RequiredArgsConstructor
public class TransDownMsgHandlerImpl implements TransDownMsgHandler {

    private final PacketFactory packetFactory;

    private final DownModeProperty downModeProperty;

    private final DownMsgSender mqSender;

    private final UserService userService;

    @Override
    public void dispatchMsg(UpMsgWrap wrap, List<GroupMemberDTO> allMembers,
                            UserWrap sender, UserWrap receiver,
                            Message message) throws Exception {
        TimeInterval timer = DateUtil.timer();
        log.info("[DISPATCHER] 转发器开始work. sender={}, messageId={}, clientMsgId={}",
            sender.getUserId(), message.getId(), message.getClientMsgId());
        message.setFlow(MessageFlowEnum.DOWN_MSG.getCode());

        handleSender(wrap, sender, message);

        handleReceiver(wrap, allMembers, sender, receiver, message);

        if (timer.interval() > 500) {
            log.warn("[DISPATCHER] 转发消息消耗过长，messageId={}, clientId={}, use:{}ms", message.getId(),
                message.getClientMsgId(), timer.interval());
        }
    }

    private void handleReceiver(UpMsgWrap wrap, List<GroupMemberDTO> allMembers, UserWrap sender,
                                UserWrap receiver, Message message) throws Exception {
        switch (message.getConversationType()) {
            case "c2c":
                forwardMsg2Recv(message, receiver);
                break;
            case "c2g":
                // 其他群成员
                forwardMsg2GrpMemberClients(wrap, allMembers, sender, message);
                break;
            default:
                log.error("[dispatch] 坏了，不知道转哪了. msgId={}", message.getId());
        }
    }

    private void forwardMsg2Recv(Message msg, UserWrap toUser) {
        // 转发给to 在线设备msg
        List<UserDevice> toOnlineClients = userService.getOnlineDevicesByUserId(toUser.getAppKey(), toUser.getUserId());
        sendRecvUserOnlineClients(msg, toOnlineClients, toUser);
    }

    private void sendRecvUserOnlineClients(Message message, List<UserDevice> toOnlineClients, UserWrap toUser) {
        message.setType(MsgTypeEnum.DOWN_MSG_C2C_RECEIVER_MSG.getCode());
        Packet packet = packetFactory.newDownmsgSuccPacket(DownCMDEnum.C2C.getCode(), message);

        // 用户在线发送消息
        if (CollUtil.isNotEmpty(toOnlineClients)) {
            log.info("[c2c] 开始向用户【{}】的在线客户端, 转发消息！", toUser.getNick());
            toOnlineClients.stream().forEach(item -> {

                DownMsgRequest request = new DownMsgRequest();
                request.setMessageId(message.getId());
                request.setIdentityNo(IdentityUtil.buildIdentityNo(toUser.getUserId(), item.getDeviceType()));
                request.setPacket(packet);
                log.info("[c2c] send msg client:{}, id:{}", request.getIdentityNo(), message.getId());

                mqSender.syncSendByMq(DownTopicConst.FLUSH_ONE_TOPIC, item.getMqTags(), JsonUtils.toJsonString(request));
//                    ForwardPacketUtil.flushPacket(request, item.getDubboIp(), item.getDubboPort());
                if (null != message.getId() && !QosSendDaemonS2C.getInstance().exist(message.getId())) {
                    QosSendDaemonS2C.getInstance().put(genQosTransBo(message.getId(), packet, toUser.getAppKey(),
                        request.getIdentityNo(), item.getMqTags()));
                }

            });
            return;
        }
        log.info("[c2c] 用户不在线, 已保存离线消息。id:{}:{}", message.getId(), message.getPayload());
    }

    @NotNull
    private PacketTransBO genQosTransBo(String msgId, Packet packet, String appkey, String identityNo, String mqTags) {
        PacketTransBO packetTransBO = new PacketTransBO();
        packetTransBO.setPacket(packet);
        packetTransBO.setQos(true);
        packetTransBO.setIdentityNo(identityNo);
        packetTransBO.setMessageId(msgId);
        packetTransBO.setAppKey(appkey);

        packetTransBO.setDownMsgMode(downModeProperty.getMode());
        packetTransBO.setMqTags(mqTags);

        packetTransBO.setIfGroupMsg(false);
        return packetTransBO;
    }

    private void handleSender(UpMsgWrap wrap, UserWrap sender, Message message) throws Exception {
        // 转发我自己
        replyAck2CurrDevice(IdentityUtil.buildIdentityNo(wrap.getBindingUserId(), wrap.getDeviceType()), wrap, message);

        syncMsg2SenderClients(wrap, sender, message);
    }

    /**
     * 回复发送消息的设备， ack
     * 系统已收到消息。客户端可以停止qos了
     *
     * @param wrap
     * @param message
     */
    private void replyAck2CurrDevice(final String identityNo, final UpMsgWrap wrap, Message message) throws Exception {
        // 获取自己信息，判断在不在，不在不发消息！
        // TODO

        QoSReceiveDaemonC2S.getInstance().addReceived(genUserMsgFingerChar(IdentityUtil.buildIdentityNo(message.getSenderId(),
            wrap.getDeviceType()), message.getClientMsgId()));

        DownMsgRequest request = new DownMsgRequest();
        request.setIdentityNo(identityNo);
        Message reply = new Message();
        reply.setSenderId(message.getSenderId()); // 系统发来的消息
        reply.setReceiverId(message.getReceiverId());
        reply.setType(MsgTypeEnum.C2G_SENDER_ACK.getCode());
        reply.setClientMsgId(message.getClientMsgId());
        reply.setId(message.getId());
        reply.setCid(message.getReceiverId());
        reply.setConversationType(ConverTypeEnum.C2G.getCode());
        reply.setPayload(message.getPayload());
        reply.setPayloadType(message.getPayloadType());

        request.setPacket(packetFactory.newDownmsgSuccPacket(DownCMDEnum.C2G.getCode(), reply));
        request.setMessageId(message.getId());

//        mqSender.syncSendByMq(DownTopicConst.FLUSH_ONE_TOPIC, wrap.getMqTags(), JSONUtil.toJsonStr(request));
        DownModeConfig config = new DownModeConfig();
        config.setMode(downModeProperty.getMode());
        config.setMqTags(wrap.getMqTags());
        config.setDubboIp(wrap.getDubboIp());
        config.setDubboPort(wrap.getDubboPort());
        mqSender.flushOne(request, config);
        log.info("[DISPATCHER] 转发给发送设备ack完毕。identityNo={}", identityNo);
//        ForwardPacketUtil.flushPacket(request, wrap.getDubboIp(), wrap.getDubboPort());

    }

    @NotNull
    protected PacketTransBO genQosTransBo(Packet packet, String identityNo, String msgId,
                                          String appId, String mqTags, String dubboIp, int dubboPort) {
        PacketTransBO packetTransBO = new PacketTransBO();
        packetTransBO.setPacket(packet);
        packetTransBO.setQos(true);
        packetTransBO.setIdentityNo(identityNo);
        packetTransBO.setMessageId(msgId);
        packetTransBO.setAppKey(appId);

        packetTransBO.setDownMsgMode(downModeProperty.getMode());
        packetTransBO.setMqTags(mqTags);

        packetTransBO.setDubboIp(dubboIp);
        packetTransBO.setDubboPort(dubboPort);
        packetTransBO.setIfGroupMsg(true);
        return packetTransBO;
    }

    /**
     * 生成消息指纹
     *
     * @param userNo
     * @param clientMsgId
     * @return
     */
    @NotNull
    protected static String genUserMsgFingerChar(String userNo, String clientMsgId) {
        return userNo + COLON + clientMsgId;
    }

    private void syncMsg2SenderClients(UpMsgWrap wrap, UserWrap sender, Message message) throws Exception {
        message.setType(MsgTypeEnum.DOWN_MSG_C2G_SENDER_SYNC.getCode());
        Packet syncPacket = packetFactory.newDownmsgSuccPacket(DownCMDEnum.C2G.getCode(), message);

        // 获取from其他在线客户端
        // 转为获取userCache 数据。没有取redis TODO
        List<UserDevice> fromOnlineClients = userService.getOnlineDevicesExceptThisDevice(sender.getAppKey(), sender.getUserId(), wrap.getDeviceType());
//        List<UserDeviceInfo> fromOnlineClients = cacheService.getOnlineClientForUserNo(sender.getAppKey(), sender.getUserId(), wrap.getDeviceType());
        // 向from其他在线客户端同步消息
        if (CollectionUtil.isNotEmpty(fromOnlineClients)) {
            for (UserDevice client : fromOnlineClients) {
                log.info("[c2g] 消息同步 client:{}", DeviceEnum.getDeviceNameByValue(client.getDeviceType()));
                DownMsgRequest request = new DownMsgRequest();
                request.setPacket(syncPacket);
                request.setMessageId(message.getId());
                request.setIdentityNo(IdentityUtil.buildIdentityNo(sender.getUserId(), client.getDeviceType()));

                //ForwardPacketUtil.flushPacket(request, client.getDubboIp(), client.getDubboPort());
//                mqSender.syncSendByMq(DownTopicConst.FLUSH_ONE_TOPIC, wrap.getMqTags(), JSONUtil.toJsonStr(request));
                DownModeConfig config = new DownModeConfig();
                config.setMode(downModeProperty.getMode());
                config.setMqTags(wrap.getMqTags());
                config.setDubboIp(wrap.getDubboIp());
                config.setDubboPort(wrap.getDubboPort());
                mqSender.flushOne(request, config);
                log.info("[DISPATCHER] 转发发送者其他客户端。identityNo={}", request.getIdentityNo());

                QosSendDaemonS2C.getInstance()
                    .put(genQosTransBo(syncPacket, request.getIdentityNo(), message.getId(),
                        sender.getAppKey(), client.getMqTags(), client.getDubboIp(), client.getDubboPort()));
            }

        }
    }

    /**
     * 这块可以使用线程池接管 分发
     * 加上队列，就可以按照不同群成员等级，实现不同分发速度。比如单聊优先级高于群，vip群高于普通群，vip小群高与vip大群，
     *
     * @param allMembers
     * @param sender
     * @param message
     */
    private void forwardMsg2GrpMemberClients(UpMsgWrap wrap, List<GroupMemberDTO> allMembers, UserWrap sender, Message message) throws Exception {
        // 返回包
        message.setType(MsgTypeEnum.DOWN_MSG_C2G_RECEIVER_.getCode());
        Packet recvPacket = packetFactory.newDownmsgSuccPacket(DownCMDEnum.C2G.getCode(), message);

        // TODO 拆分500人群，3000人群， 100000人群
        // date: 2024-12-15

        List<String> users = allMembers.stream().map(item -> item.getUserId()).collect(Collectors.toList());

        Map<String, List<UserDeviceDetailInfo>> map = userService.obtainOnlineDevicesByClassifyMqTags(sender.getAppKey(), users);
//        Map<String, List<UserSingleDeviceInfo>> map = cacheService.classifyOnlineUsersByMqTags(sender.getAppKey(), users);

        for (Map.Entry<String, List<UserDeviceDetailInfo>> entry : map.entrySet()) {
            List<UserDeviceDetailInfo> value = entry.getValue();
            if (CollectionUtil.isEmpty(value)) {
                continue;
            }
            List<String> identityNos = value.stream()
                .map(item -> IdentityUtil.buildIdentityNo(item.getUserId(), item.getDeviceType()))
                .collect(Collectors.toList());

            UserDeviceDetailInfo userInfo = value.get(0);

            DownMsgBatchReq batchReq = new DownMsgBatchReq();
            batchReq.setIdentityNos(identityNos);
            batchReq.setPacket(recvPacket);

//            mqSender.syncSendByMq(DownTopicConst.FLUSH_MORE_TOPIC, userInfo.getMqTags(), JSONUtil.toJsonStr(batchReq));
//            ForwardPacketUtil.flushMore(batchReq, userInfo.getDubboIp(), userInfo.getDubboPort());
            DownModeConfig config = new DownModeConfig();
            config.setMode(downModeProperty.getMode());
            // 向 分组的 mqTags发送， 不是自己的tags
            config.setMqTags(entry.getKey());
            // 没有用准备删除
            config.setDubboIp(wrap.getDubboIp());
            config.setDubboPort(wrap.getDubboPort());
            mqSender.flushMore(batchReq, config);

            Map<String, UserDeviceDetailInfo> allUserMap = value.stream().distinct().collect(Collectors.toMap(key -> IdentityUtil.buildIdentityNo(key.getUserId(), key.getDeviceType()), Function.identity()));
            for (String identityNo : identityNos) {
                UserDeviceDetailInfo l1User = allUserMap.get(identityNo);
                if (null == l1User) {
                    continue;
                }
                QosSendDaemonS2C.getInstance().put(genQosTransBo(recvPacket, identityNo, message.getId(),
                    l1User.getAppKey(), l1User.getMqTags(), l1User.getDubboIp(), l1User.getDubboPort()));
            }

        }

    }

}
